package com.icarbonx.dap.notification.job

import com.google.gson.Gson
import com.icarbonx.dap.notification.msgservice.rabbitmq.{ConnUtil, MQObject}
import com.icarbonx.dap.notification.util.{DateUtil, KafkaUtil, NoticeMsg, SparkUtil}
import com.icarbonx.dap.notification.vo._
import kafka.serializer.StringDecoder
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

/**
 * Created by Lenovo on 2018/8/12.
 * 19.多发午餐前低糖警报:连续两个点血糖<3.0mmol/L,5天内出现3次&既定午餐时间前1小时内:low_glycemia_lunch_freq:每天午餐前:触发及时发送，每小时提示不超过1条
 * 20.多发晚餐前低糖警报:连续两个点血糖<3.0mmol/L,5天内出现3次&既定晚餐时间前2小时内:low_glycemia_supper_freq:每天晚餐前:触发及时发送，每小时提示不超过1条
 * 21.多发餐后高糖警报:连续两个点血糖>10.1mmol/L,5天内出现3次&进餐后1小时内:high_glycemia_post_meal_freq:每天晚餐后:触发及时发送，每小时提示不超过1条
 * 25.午餐前低糖:连续两个点血糖<2.8mmol/L&既定午餐时间前1小时内:low_glycemia_lunch:每次午餐前:触发及时发送
 * 26.晚餐前低糖:连续两个点血糖<2.8mmol/L&既定晚餐时间前2小时内:low_glycemia_supper:每次晚餐前:触发及时发送
 * 29.餐后高糖:连续2个点血糖>12.0mmol/L&进餐后1小时内:high_glycemia_meal:每次用餐后:触发及时发送
 * 31.餐后升糖过速:血糖上升变化超过6.0mmol/L/h&用餐后1小时内:slope_glycemia_meal:每次用餐后:触发及时发送
 * 总计7个警报
 */
object BloodSugarMealStream {
  def main(args: Array[String]) {
    Logger.getLogger("org.apache.kafka").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    /**
     * brokers = "10.47.250.4:9092,10.47.250.7:9092,10.47.250.6:9092"
     * topicList = "abel_test"
     * mq:10.104.103.7 5672 guest icx@betaOn16 /
     * redis:10.104.103.7 6379 passxx#$#dslllfjshwerw123@ 3000 abel_test
     */
    println(args.mkString("\t"))
    val Array(brokers, topicList, mealTopic, foodTopic, mqHost, mqPort, mqUsername,
    mqPassword, mqVirtualHost, exchangeName, mqRouteKey, checkpointDir, duation, rddDuration, foodPartition, appName) = args
    val conf = SparkUtil.getSparkConf(appName, Array(classOf[BloodSugarRecord], classOf[FoodRecord], classOf[MealRecord]))

    val sc = SparkContext.getOrCreate(conf)

    //    val checkpointDir = "/data/streaming/checkpoint/bloodsugar_meal_stream/"

    val updateSugarFunc = (values: Seq[ArrayBuffer[BloodSugarRecord]],
                           state: Option[ArrayBuffer[BloodSugarRecord]]) => {
      //保存状态
      val HourTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 8 * 60 * 60 * 1000
      val currentCount = ArrayBuffer[BloodSugarRecord]()
      values.foreach(list => currentCount ++= list)
      val previousCount = state.getOrElse(ArrayBuffer[BloodSugarRecord]())
      Some(currentCount ++ previousCount.filter(_.timestamp > HourTimeStamp))
    }
    val updateMealFunc = (values: Seq[ArrayBuffer[Long]],
                          state: Option[ArrayBuffer[Long]]) => {
      //保存状态
      val HourTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 8 * 60 * 60 * 1000
      val currentCount = ArrayBuffer[Long]()
      values.foreach(list => currentCount ++= list)
      val previousCount = state.getOrElse(ArrayBuffer[Long]())
      Some(currentCount ++ previousCount.filter(_ > HourTimeStamp))
    }

    val updateMealTimeStateFunc =
      (values: Seq[mutable.HashMap[Long, Int]],
       state: Option[mutable.HashMap[Long, Int]]) => {
        //保存状态
        val fiveDaysAgoTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 5 * 24 * 60 * 60 * 1000
        val currentState = mutable.HashMap[Long, Int]()
        val previousState = state.getOrElse(mutable.HashMap[Long, Int]())
        values.foreach {
          case map =>
            map.foreach {
              case (mealTime, flag) =>
                currentState += ((mealTime, flag))
            }
        }
        val tempOldState = previousState.filter {
          case (mealTime, flag) =>
            !currentState.contains(mealTime) && mealTime > fiveDaysAgoTimeStamp
        }
        Some(currentState ++= tempOldState)
      }

    def creatingFunc(): StreamingContext = {
      val ssc = new StreamingContext(sc, Seconds(duation.toInt))
      ssc.checkpoint(checkpointDir)
      val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

      val mqConn = new MQObject(mqHost, mqPort.toInt, mqUsername, mqPassword, mqVirtualHost)

      val sugarTopics = topicList.split(",").toSet
      val mealTopics = mealTopic.split(",").toSet
      val foodTopics = foodTopic.split(",").toSet

      val sugarMessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder]( ssc, kafkaParams, sugarTopics )
      val mealMessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder]( ssc, kafkaParams,mealTopics )
      val foodMessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder]( ssc, kafkaParams,foodTopics )

      val sugarBatch: DStream[(String, ArrayBuffer[BloodSugarRecord])] =
        sugarMessages
          .filter(_._2.nonEmpty)
          .flatMap { msg =>
            try {
              println("___________ msg :" + msg._2)
              val gson = new Gson()
              val voList =
                gson.fromJson(msg._2, classOf[Array[BloodSugarRecord]])
              voList.map { vo =>
                (vo.personId.toString, ArrayBuffer[BloodSugarRecord](vo))
              }.toList
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                List(("", ArrayBuffer[BloodSugarRecord]()))
            }
          }
          .filter(_._1.nonEmpty)
          .updateStateByKey(updateSugarFunc)

      val mealBatch: DStream[(String, ArrayBuffer[Long])] = mealMessages
        .filter(_._2.nonEmpty)
        .map { msg =>
          try {
            println("___________ msg :" + msg)
            val gson = new Gson()
            val vo = gson.fromJson(msg._2, classOf[MealRecord])
            (vo.personId.toString, ArrayBuffer[Long](vo.recordTime))
          } catch {
            case e: Exception =>
              println("____________" + e.getMessage)
              ("", ArrayBuffer[Long]())
          }
        }
        .filter(_._1.nonEmpty)

      val foodBatch: DStream[(String, ArrayBuffer[Long])] = foodMessages
        .filter(_._2.nonEmpty)
        .map { msg =>
          try {
            println("___________ msg :" + msg)
            val gson = new Gson()
            val vo = gson.fromJson(msg._2, classOf[FoodRecord])
            (vo.personId.toString, ArrayBuffer[Long](vo.ateTime))
          } catch {
            case e: Exception =>
              println("____________" + e.getMessage)
              ("", ArrayBuffer[Long]())
          }
        }
        .filter(_._1.nonEmpty)

      val foodAndMeal = mealBatch.union(foodBatch).repartition(foodPartition.toInt).updateStateByKey(updateMealFunc)

      val notNullSugar = sugarBatch.filter(_._2.nonEmpty)
      val notNullFood = foodAndMeal.filter(_._2.nonEmpty)

      notNullSugar.checkpoint(Duration(rddDuration.toInt * 1000))
      notNullFood.checkpoint(Duration(rddDuration.toInt * 1000))

      val sugarAndMeal = notNullFood.join(notNullSugar)
      val sortRdd = sugarAndMeal
        .filter {
          case (pid, (mealList, sugarList)) =>
            mealList.nonEmpty && sugarList.nonEmpty
        }
        .map {
          case (pid, (mealList, sugarList)) =>
            (pid, (mealList.distinct.sortWith(_.compareTo(_) < 0), sugarList.sortBy(_.timestamp)))
        }

      //@todo （多发餐后高糖记录）
      val mealTimeDoneFlag = sortRdd
        .map {
          case (pid, (mealList, sugarList)) =>
            println("____ pid :" + pid + "____ mealList :" + mealList.toString() + "____ sugarList :" + sugarList.toString())
            val result = mutable.HashMap[Long, Int]()
            mealList.foreach { recordTime =>
              val afterMealOneHour = recordTime + 60 * 60 * 1000
              val afterRangeTimeSugar = sugarList.filter(
                sugarVo =>
                  sugarVo.timestamp <= afterMealOneHour && sugarVo.timestamp > recordTime
              )
              if (afterRangeTimeSugar.size >= 2) {
                var preVo = afterRangeTimeSugar.head
                for (i <- 1 to afterRangeTimeSugar.size - 1) {
                  val nowVo = afterRangeTimeSugar(i)
                  if (nowVo.timestamp - preVo.timestamp < 15 * 60 * 1000
                    && preVo.value > 10.1 && nowVo.value > 10.1) {
                    result += ((DateUtil.getDateTimestamp(DateUtil.getTimeStampFormat(recordTime)), 1))
                  }
                  preVo = nowVo
                }
              }
            }
            (pid, result)
        }
        .filter(_._2.nonEmpty)
        .updateStateByKey(updateMealTimeStateFunc)

      mealTimeDoneFlag.checkpoint(Duration(rddDuration.toInt * 1000))

      //@todo 21.多发餐后高糖警报:连续两个点血糖>10.1mmol/L,5天内出现3次&进餐后1小时内:high_glycemia_post_meal_freq:每天晚餐后:触发及时发送，每小时提示不超过1条
      mealTimeDoneFlag.foreachRDD { mess =>
        if (!mess.isEmpty()) {
          mess.foreachPartition { lines =>
            val mqMsg = ArrayBuffer[String]()
            val gson = new Gson()
            lines.foreach {
              case (pid, mealListState) =>
                println("____ pid :" + pid + "____ mealListState :" + mealListState.toString())
                if (mealListState.values.sum >= 3) {
                  val msg = new WarningMsg(
                    "glycemia",
                    pid.toLong,
                    NoticeMsg.high_glycemia_post_meal_freq._1,
                    NoticeMsg.high_glycemia_post_meal_freq._2,
                    NoticeMsg.high_glycemia_post_meal_freq._3,
                    mealListState.keys.max.toString,
                    NoticeMsg.high_glycemia_post_meal_freq._5,
                    NoticeMsg.high_glycemia_post_meal_freq._4,
                    NoticeMsg.high_glycemia_post_meal_freq._6
                  )
                  mqMsg += gson.toJson(msg)
                }
            }
            if (mqMsg.nonEmpty) {
              val mqChannel = mqConn.connect().createChannel()
              mqMsg.foreach { msg =>
                ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
              }
              mqChannel.close()
            }
          }
        }
      }

      sortRdd.foreachRDD(mess =>
        if (!mess.isEmpty) {
          mess.foreachPartition(lines => {
            val mqMsg = ArrayBuffer[String]()
            val gson = new Gson()
            lines.foreach {
              case (pid, (mealList, sugarList)) => {
                println("____ pid :" + pid + "____ mealList:" + mealList.map(gson.toJson(_)).toString()
                  + "____ sugarList" + sugarList.map(gson.toJson(_)).toString()
                )

                mealList.foreach {
                  recordTime =>
                    //@todo  29.餐后高糖:连续2个点血糖>12.0mmol/L&进餐后1小时内:high_glycemia_meal:每次用餐后:触发及时发送
                    val afterMealOneHour = recordTime + 60 * 60 * 1000
                    val rangeTimeSugar = sugarList.filter(
                      sugarVo =>
                        sugarVo.timestamp <= afterMealOneHour && sugarVo.timestamp >= recordTime
                    )
                    if (rangeTimeSugar.size >= 2) {
                      var preVo = rangeTimeSugar.head
                      for (i <- 1 to rangeTimeSugar.size - 1) {
                        val nowVo = rangeTimeSugar(i)
                        if (preVo.value > 12 && nowVo.value > 12) {
                          val msg = new WarningMsg(
                            "glycemia",
                            pid.toLong,
                            NoticeMsg.high_glycemia_meal._1,
                            NoticeMsg.high_glycemia_meal._2,
                            NoticeMsg.high_glycemia_meal._3,
                            recordTime.toString,
                            NoticeMsg.high_glycemia_meal._5,
                            NoticeMsg.high_glycemia_meal._4,
                            NoticeMsg.high_glycemia_meal._6
                          )
                          mqMsg += gson.toJson(msg)
                        }
                        preVo = nowVo
                      }
                    }

                    //@todo 31.餐后升糖过速:血糖上升变化超过6.0mmol/L/h&用餐后1小时内:slope_glycemia_meal:每次用餐后:触发及时发送
                    val inOneHourRangeTimeSugar = sugarList.filter(
                      sugarVo =>
                        sugarVo.timestamp <= afterMealOneHour && sugarVo.timestamp > recordTime
                    )
                    inOneHourRangeTimeSugar.foreach {
                      vo =>
                        val beforeThisOneHour = vo.timestamp - 60 * 60 * 1000
                        val beforeThisOneListMinSugar = sugarList
                          .filter(
                            sugarVo =>
                              sugarVo.timestamp < vo.timestamp && sugarVo.timestamp >= beforeThisOneHour
                          )
                        if (beforeThisOneListMinSugar.nonEmpty) {
                          if (vo.value - beforeThisOneListMinSugar.minBy(_.value).value > 6) {
                            val msg = new WarningMsg(
                              "glycemia",
                              pid.toLong,
                              NoticeMsg.slope_glycemia_meal._1,
                              NoticeMsg.slope_glycemia_meal._2,
                              NoticeMsg.slope_glycemia_meal._3,
                              recordTime.toString,
                              NoticeMsg.slope_glycemia_meal._5,
                              NoticeMsg.slope_glycemia_meal._4,
                              NoticeMsg.slope_glycemia_meal._6
                            )
                            mqMsg += gson.toJson(msg)
                          }
                        }
                    }
                }
              }
            }
            if (mqMsg.nonEmpty) {
              val mqChannel = mqConn.connect().createChannel()
              mqMsg.foreach { msg =>
                ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
              }
              mqChannel.close()
            }
          })
        })
      ssc
    }


    val ssc = StreamingContext.getOrCreate(checkpointDir, creatingFunc)

    ssc.start()
    ssc.awaitTermination()
  }
}